package com.dyna.cp.kafka.controller;

import com.dyna.cp.kafka.service.KafkaService;
import com.dyna.domain.resp.R;
import com.dyna.domain.resp.ResultCode;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;

/**
 * demo模块/kafka相关
 * @author zhoucaiwang
 * @date 2024/5/14
 */
@Api(value = "demo模块/cp/kafka相关")
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    private final KafkaService kafkaService;

    @Autowired
    public KafkaController(KafkaService kafkaService) {
        this.kafkaService = kafkaService;
    }

    @ApiOperation(value = "kafka测试")
    @RequestMapping("/test-kafka")
    public R<Boolean> testKafka(@RequestParam(value = "msg") String msg) {
        kafkaService.testKafka(msg);
        return R.isSuccess(Boolean.TRUE);
    }

    @ApiOperation(value = "测试回滚kafka消息是否通知成功")
    @RequestMapping("/kafka-rollback")
    public R<Boolean> kafkaRollback(@RequestParam(value = "msg") String msg) {
        kafkaService.kafkaRollback(msg);
        return R.isSuccess(Boolean.TRUE);
    }

    @ApiOperation("测试kafka消息读取顺序")
    @RequestMapping("/order/consumer/msg")
    public R<Object> orderConsumerMsg() {
        kafkaService.orderConsumerMsg();
        return R.success(ResultCode.SUCCESS);
    }

    @ApiOperation("测试kafka发送到指定分区")
    @GetMapping("/sendDesignatePartition")
    public R<Boolean> sendDesignatePartition(@RequestParam("topic") String topic, @RequestParam("partition") Integer partition, @RequestParam("key") String key, @RequestParam("msg") String msg) {
        kafkaService.sendDesignatePartition(topic, partition, key, msg);
        return R.isSuccess(Boolean.TRUE);
    }

    @ApiOperation("批量生产消息")
    @GetMapping("/batch-product-msg")
    public R<Boolean> batchProductMsg(@RequestBody List<String> msgList) {
        kafkaService.batchProductMsg(msgList);
        return R.data(Boolean.TRUE);
    }

}